package me.seawenc.datastash.inputs.kafka;

import me.seawenc.datastash.DatastashApplication;
import me.seawenc.datastash.inputs.AbstractProducer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.util.Arrays;
import java.util.Properties;
import java.util.function.Consumer;

/**
 * kafka消息消费者测试
 */
public class KafkaProducer extends AbstractProducer {
    public KafkaProducer(Properties fileProps) {
        super(fileProps);
        this.props = getKafkaConfig(fileProps);
        System.out.println("kafka完整参数：");
        this.props.forEach((k,v)-> System.out.println(k+"="+v));
    }

    private Properties getKafkaConfig(Properties fileProps) {
        Properties properties = new Properties();
        fileProps.stringPropertyNames().stream()
                .filter(key-> key.startsWith("input.kafka."))
                .forEach(key-> properties.put(key.replace("input.kafka.",""),fileProps.getProperty(key)));

        setDefValue(properties,"acks","all");
        setDefValue(properties,"key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        setDefValue(properties,"value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        setDefValue(properties,"key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        setDefValue(properties,"value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        setDefValue(properties,"max.request.size","10485760");
        setDefValue(properties,"auto.offset.reset","earliest");

        return properties;
    }

    @Override
    public void producer(Consumer<String> consumer) throws Exception {
        while (DatastashApplication.isRunning){
            try {
                run(consumer);
            } catch (Exception e) {
                System.out.println("kafka接数异常，msg:"+e.getMessage());
                System.out.println("kafka已断线，3秒后后自动重连接...");
                Thread.sleep(3000);
            }
        }
    }

    private void run(Consumer<String> consumer) throws Exception {
        while (true) {
            org.apache.kafka.clients.consumer.KafkaConsumer<String, String> kafkaConsumer = new org.apache.kafka.clients.consumer.KafkaConsumer<>(props);
            kafkaConsumer.subscribe(Arrays.asList(props.getProperty("topic")));
            while (true) {
                Thread.sleep(100);
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    String value = record.value();
                    send(consumer,value);
                }
            }
        }
    }

    private void setDefValue(Properties target, String key, String def){
        String val = target.getProperty(key);
        if(val==null || "".equals(val)){
            target.setProperty(key,def);
        }
    }
}